package com.spider.moudle.task;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.spider.moudle.api.Writer;
import com.spider.moudle.api.realize.StandardLogRecorder;
import com.spider.moudle.constant.Constant;
import com.spider.moudle.entity.*;
import com.spider.moudle.pool.TaskQueryPool;
import com.spider.moudle.pool.WriterPool;
import com.spider.moudle.util.CommonUtil;
import com.spider.moudle.util.DateUtil;
import com.spider.moudle.util.ErrorLogUtil;
import com.spider.moudle.util.PropertiesUtil;

import java.util.*;
import java.util.concurrent.*;

/**
 * Created by Linbo Hu on 2017/11/15.
 * 任务管理器
 */
public class TaskManager {

    private static final Map<String, BlockingQueue<Task>> taskQueueMap = Maps.newConcurrentMap();

    private static WriterPool writerPool = WriterPool.getInstance();

    private static TaskQueryPool taskQueryPool = TaskQueryPool.getInstance();

    private static String runMethod = PropertiesUtil.getVal("run.method");

    private static String taskPushWay = PropertiesUtil.getVal("task.push.way");

    private static StandardLogRecorder standardLogRecorder = StandardLogRecorder.getInstance();

    /**
     * 添加任务
     *
     * @param taskName 任务名称 (也是具体的某个网站的任务名称)
     * @param task     具体的任务
     * @throws InterruptedException
     */
    public static void add(String taskName, Task task) throws Exception {
        if (task.getTaskType() != null && TaskType.USER == task.getTaskType()) {
            String userKey = task.getTaskKey();
            if (CommonUtil.isNull(userKey)) {
                return;
            }
            TaskQuery taskQuery = null;
            try {
                taskQuery = taskQueryPool.fetchTaskQuery();
                long start = System.currentTimeMillis();
                Boolean isExist = taskQuery.userQuery(task);
                if (isExist != null && isExist) {
                    System.out.println("用户 " + userKey + " 查询耗时：" + (System.currentTimeMillis() - start) + "ms, 结果--存在");
                }
                if (isExist != null && isExist) {
                    return;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (null != taskQuery) {
                    taskQueryPool.releaseTaskQuery(taskQuery);
                }
            }
        } else if (task.getTaskType() != null && TaskType.SPEECH == task.getTaskType()) {
            TaskQuery taskQuery = null;
            try {
                taskQuery = taskQueryPool.fetchTaskQuery();
                long start = System.currentTimeMillis();
                Boolean isExist = taskQuery.speechQuery(task);
                if (isExist != null && isExist) {
                    System.out.println("言论 " + task.getUrl() + " 查询耗时：" + (System.currentTimeMillis() - start) + "ms, 结果--存在");
                }
                if (isExist != null && isExist) {
                    return;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (null != taskQuery) {
                    taskQueryPool.releaseTaskQuery(taskQuery);
                }
            }
        } else {
            System.err.println("the value of task type is unavailable");
        }
        if ("KAFKA".equals(taskPushWay)) {
            Writer writer = null;
            try {
                writer = writerPool.fetch(WriterType.KAFKA);
                writer.write(taskName, task);
                //记录此网站的任务生成量
                standardLogRecorder.addTaskCount(taskName);
            } catch (Exception e) {
                ErrorLogUtil.error(e.getMessage(), e);
            } finally {
                if (null != writer) {
                    writerPool.release(WriterType.KAFKA, writer);
                }
            }
        } else if ("LOCAL".equals(taskPushWay)) {
            addLocal(taskName, task);
            //记录此网站的任务生成量
            standardLogRecorder.addTaskCount(taskName);
            System.out.println("{name: '" + taskName + "', type: '" + task.getTaskType() + "', status:'success',  nowTime:'" + DateUtil.getNowTimeString(DateUtil.yyyy_MM_dd_HH_mm_ss) + "'}");
        } else {
            throw new Exception("The value of task.push.type must be KAFKA or LOCAL");
        }
    }

    /**
     * 将任务添加到本地任务池中
     *
     * @param taskName 任务名称
     * @param task     具体的任务
     * @throws Exception
     */
    public static void addLocal(String taskName, Task task) throws Exception {
        BlockingQueue<Task> taskQueue = taskQueueMap.get(taskName);
        if (taskQueue == null) {
            taskQueue = Queues.newLinkedBlockingQueue();
            if ("DEVELOP".equals(runMethod)) {
                System.out.println("taskQueueMap的大小:" + taskQueueMap.size() + "-----put:" + taskName + "-" + taskQueue.size());
            }
            taskQueue.put(task);
            synchronized (taskQueueMap) {
                taskQueueMap.put(taskName, taskQueue);
                taskQueueMap.notifyAll();
            }
        } else {
            if ("DEVELOP".equals(runMethod)) {
                System.out.println("taskQueueMap的大小:" + taskQueueMap.size() + "-----put:" + taskName + "-" + taskQueue.size());
            }
            switch (PropertiesUtil.getVal("spider.work.mode")){
                case "SPEECH":{
                    if(taskQueueMap.entrySet().size()>1&&taskQueue.size()>Constant.MAX_SPEECH_TASK_CACHE_SIZE/2){
                        if ("DEVELOP".equals(runMethod)) {
                            System.out.println("丢弃任务:"+task);
                        }
                        return;
                    }
                    break;
                }
                case "USER":{
                    if(taskQueueMap.entrySet().size()>1&&taskQueue.size()>Constant.MAX_USER_TASK_CACHE_SIZE/2){
                        if ("DEVELOP".equals(runMethod)) {
                            System.out.println("丢弃任务:"+task);
                        }
                        return;
                    }
                    break;
                }
                case "BOTH":{
                    if(task.getTaskType() == TaskType.USER){
                        if(taskQueue.size()>Constant.MAX_USER_TASK_CACHE_SIZE/2){
                            if ("DEVELOP".equals(runMethod)) {
                                System.out.println("丢弃任务:"+task);
                            }
                            return;
                        }
                    }else if(task.getTaskType() == TaskType.SPEECH){
                        if(taskQueue.size()>Constant.MAX_SPEECH_TASK_CACHE_SIZE/2){
                            if ("DEVELOP".equals(runMethod)) {
                                System.out.println("丢弃任务:"+task);
                            }
                            return;
                        }
                    }
                    break;
                }
                default:{
                    System.err.println("the value of spider.work.mode is unavailable");
                }
            }
            taskQueue.put(task);
        }
    }


    /**
     * 获取一个任务,非阻塞方法
     *
     * @param taskName 任务名称
     * @param mills    超时时间
     * @return
     * @throws InterruptedException
     */
    public static Task takeTask(String taskName, long mills) throws Exception {
        BlockingQueue<Task> taskQueue = taskQueueMap.get(taskName);
        if (taskQueue == null) {
            taskQueue = Queues.newLinkedBlockingQueue();
            synchronized (taskQueueMap) {
                taskQueueMap.put(taskName, taskQueue);
                taskQueueMap.notifyAll();
            }
        }
        if (taskQueue.size() > 0) {
            if ("DEVELOP".equals(runMethod)) {
                System.out.println("taskQueueMap的大小:" + taskQueueMap.size() + "-----take:" + taskName + "-" + taskQueue.size());
            }
        }
        return taskQueue.poll(mills, TimeUnit.MILLISECONDS);
    }

    /**
     * 根据任务队列名称获取某个队列的长度
     *
     * @param name
     * @return
     */
    public static Integer getCount(String name) {
        BlockingQueue<Task> taskQueue = taskQueueMap.get(name);
        if (taskQueue == null) {
            return null;
        }
        return taskQueue.size();
    }

    /**
     * 将字符串解析成Task
     *
     * @param text
     * @return
     */
    public static Task parse(String text) {
        Task task = new Task();
        JSONObject object = JSON.parseObject(text);
        if (object.containsKey("url")) {
            task.setUrl(object.getString("url"));
        }
        if (object.containsKey("taskType")) {
            String taskType = object.getString("taskType");
            if ("USER".equals(taskType)) {
                task.setTaskType(TaskType.USER);
            } else if ("SPEECH".equals(taskType)) {
                task.setTaskType(TaskType.SPEECH);
            }
        }
        if (object.containsKey("webType")) {
            String webType = object.getString("webType");
            if ("FORLUM".equals(webType)) {
                task.setWebType(WebType.FORLUM);
            } else if ("NEWS".equals(webType)) {
                task.setWebType(WebType.NEWS);
            } else if ("WEIBO".equals(webType)) {
                task.setWebType(WebType.WEIBO);
            } else if ("WEIXIN".equals(webType)) {
                task.setWebType(WebType.WEIXIN);
            } else if ("QQZONE".equals(webType)) {
                task.setWebType(WebType.QQZONE);
            }
        }
        if (object.containsKey("taskKey")) {
            task.setTaskKey(object.getString("taskKey"));
        }

        if (object.containsKey("area")) {
            JSONObject area_object = object.getJSONObject("area");
            Area area = new Area();
            if (area_object.containsKey("country")) {
                area.setCountry(area_object.getString("country"));
            }
            if (area_object.containsKey("province")) {
                area.setProvince(area_object.getString("province"));
            }
            if (area_object.containsKey("city")) {
                area.setCity(area_object.getString("city"));
            }
            if (area_object.containsKey("lat")) {
                area.setLat(area_object.getString("lat"));
            }
            if (area_object.containsKey("lon")) {
                area.setLon(area_object.getString("lon"));
            }
            task.setArea(area);
        }
        if (object.containsKey("createTime")) {
            task.setCreateTime(object.getDate("createTime"));
        }
        if (object.containsKey("parameter")) {
            JSONObject parameter_object = object.getJSONObject("parameter");
            Map<String, Object> map = new HashMap<>();
            for (Map.Entry<String, Object> entry : parameter_object.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();
                map.put(key, value);
            }
            task.setParameter(map);
        }
        return task;
    }

    public static synchronized int getTaskCount(String topic) {
        int userCount = 0;
        int speechCount = 0;
        for (Map.Entry<String, BlockingQueue<Task>> entry : taskQueueMap.entrySet()) {
            if (entry.getKey().endsWith("_SPEECH")) {
                speechCount += entry.getValue().size();
            }
            if (entry.getKey().endsWith("_USER")) {
                userCount += entry.getValue().size();
            }
        }
        if (topic.equals(PropertiesUtil.getVal("kafka.speech.task.topic"))) {
            return speechCount;
        } else if (topic.equals(PropertiesUtil.getVal("kafka.user.task.topic"))) {
            return userCount;
        } else {
            return 0;
        }
    }

    public static boolean isOverLoad(String topic) {
        if (topic.equals(PropertiesUtil.getVal("kafka.speech.task.topic"))) {
            return getTaskCount(topic) > Constant.MAX_SPEECH_TASK_CACHE_SIZE;
        } else {
            return getTaskCount(topic) > Constant.MAX_USER_TASK_CACHE_SIZE;
        }
    }
}
